/*
 * (c) Copyright 2009 Hewlett-Packard Development Company, LP
 * All rights reserved.
 * [See end of file]
 */

package com.hp.hpl.jena.clusteredtdb.pgraph;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;

import lib.NotImplemented;

import com.hp.hpl.jena.clusteredtdb.base.UtilityFunctions;
import com.hp.hpl.jena.clusteredtdb.comms.ClusterRepresentation;
import com.hp.hpl.jena.clusteredtdb.comms.NodeIdNodeProcessor;
import com.hp.hpl.jena.clusteredtdb.comms.NodeNodeIdProcessor;
import com.hp.hpl.jena.clusteredtdb.comms.NullProcessor;
import com.hp.hpl.jena.clusteredtdb.comms.RemoteVNode;
import com.hp.hpl.jena.clusteredtdb.comms.Request;
import com.hp.hpl.jena.clusteredtdb.comms.StringProcessor;
import com.hp.hpl.jena.sparql.util.Utils;
import com.hp.hpl.jena.tdb.store.NodeId;

public class NetworkedObjectFile implements BulkObjectFile {

	//RemoteVNode[] vns;
	ClusterRepresentation cr;
	NodeMask mask;
	
	public NetworkedObjectFile(ClusterRepresentation cr, NodeMask mask) {
		this.cr = cr;
		this.mask = mask;
	}

	public List<String> all() {
		//should auto-resize the NodeProcessor list once rather than many times
		StringProcessor np  = new StringProcessor();
		ArrayList<String> ar = new ArrayList<String>();
		ar.add("GETALLNODES");
		np = (StringProcessor)cr.BroadcastAllVNodesRequest(ar, np);
		return new ArrayList<String>(np.getArrayData());
	}

	public void close() {
		NullProcessor np  = new NullProcessor();
		ArrayList<String> ar = new ArrayList<String>();
		ar.add("CLOSEOBJECT");
		cr.BroadcastAllRequest(ar, np);
	}

	public String read(NodeId id) {
		RemoteVNode vnode = this.findVNode(id);
		NodeIdNodeProcessor np  = new NodeIdNodeProcessor();
		ArrayList<String> ar = new ArrayList<String>();
		String foundNode = "";
		
		ar.add("IDTONODE");
		ar.add(String.valueOf(id.getId()));
		np = (NodeIdNodeProcessor)cr.SendOneVNode(new Request(vnode, ar, false),np);
		Iterator<String> nodeIterator = np.getData().values().iterator();
		while (nodeIterator.hasNext()) {
			foundNode += nodeIterator.next() + "\n";
		}
		foundNode = foundNode.substring(0, foundNode.length() - 1);
		
		return foundNode;
	}

	public void sync(boolean force) {
		NullProcessor np  = new NullProcessor();
		ArrayList<String> ar = new ArrayList<String>();
		ar.add("SYNCOBJECT");
		ar.add(String.valueOf(force));
		cr.BroadcastAllRequest(ar, np);
	}

	public NodeId write(String str) {
		RemoteVNode vnode = this.findVNode(str);
		ArrayList<String> ar = new ArrayList<String>();
		NodeId foundNode = null;
		NodeNodeIdProcessor np  = new NodeNodeIdProcessor();
		
		ar.add("WRITENODE");
		ar.add(str);
		np = (NodeNodeIdProcessor)cr.SendOneVNode(new Request(vnode, ar, true), np);
		Iterator<NodeId> nodeIterator = np.getArrayData().iterator();
		if(nodeIterator.hasNext()) {
			foundNode = nodeIterator.next();
		}
		return foundNode;
	}
	
	public Map<String,NodeId> writeBulk(Set<String> strs) {
		NodeNodeIdProcessor nidnod = new NodeNodeIdProcessor();
		HashMap<RemoteVNode, Request> reqMap = new HashMap<RemoteVNode, Request>();
		Iterator<String> it = strs.iterator();
		while(it.hasNext()) {
			String node = it.next();
			
			RemoteVNode vnode = this.findVNode(node);
			Request r = reqMap.get(vnode); 
			if(r == null) {	
				r = new Request(vnode,false);
				r.addRequestLine("WRITENODE");
				r.addRequestLine(node);
				reqMap.put(vnode, r);
			} else {
				r.addRequestLine(node);
			}
		}
		
		cr.SendMultipleRequests(reqMap.values().iterator(), nidnod);
		return nidnod.getData();
	}

	//NB remember to distinguish carefully between vnodes and nodes
	//performs a bulk read of IDs, sorting them into buckets for each vnode
	//TODO deal with duplicates.
	public Map<NodeId, String> readBulk(Set<NodeId> ids) {
		NodeIdNodeProcessor np  = new NodeIdNodeProcessor();
		HashMap<RemoteVNode, Request> reqMap = new HashMap<RemoteVNode, Request>();
		Iterator<NodeId> it = ids.iterator();
		while(it.hasNext()) {
			NodeId nid = it.next();
			
			RemoteVNode vnode = this.findVNode(nid);
			Request r = reqMap.get(vnode); 
			if(r == null) {	
				r = new Request(vnode,false);
				r.addRequestLine("IDTONODE");
				r.addRequestLine(String.valueOf(nid.getId()));
				reqMap.put(vnode, r);
			} else {
				r.addRequestLine(String.valueOf(nid.getId()));
			}
		}
		
		cr.SendMultipleRequests(reqMap.values().iterator(), np);
		return np.getData();
	}
	
	/*
	 * Selects a vnode based on the upper masksize bits of the id
	 * TODO update ID generation such that there's no chance of those 
	 * bits being filled.
	 */
	protected RemoteVNode findVNode(NodeId key) {
		return cr.getVNodeForId(this.mask.getVNodeFromNodeId(key.getId()));
	}
	
	/*
	 * Selects a vnode on which to store a node.
	 */
	protected RemoteVNode findVNode(String str) {
		//hash the key
		int hash = UtilityFunctions.hashString(str);
		
		//find a node based on the hash
		return cr.getVNodeForId(Math.abs(hash % cr.getVNodeCount()));
	}

    @Override
    public void dump()
    { throw new NotImplemented(Utils.className(this)+".dump") ; }

}

/*
 * (c) Copyright 2009 Hewlett-Packard Development Company, LP
 * All rights reserved.
 *
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions
 * are met:
 * 1. Redistributions of source code must retain the above copyright
 *    notice, this list of conditions and the following disclaimer.
 * 2. Redistributions in binary form must reproduce the above copyright
 *    notice, this list of conditions and the following disclaimer in the
 *    documentation and/or other materials provided with the distribution.
 * 3. The name of the author may not be used to endorse or promote products
 *    derived from this software without specific prior written permission.
 *
 * THIS SOFTWARE IS PROVIDED BY THE AUTHOR ``AS IS'' AND ANY EXPRESS OR
 * IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES
 * OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED.
 * IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY DIRECT, INDIRECT,
 * INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT
 * NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
 * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
 * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
 * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF
 * THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 */
